package com.songyang.yougnsql.core.index;

import com.songyang.yougnsql.common.ExceptionHandler;
import com.songyang.yougnsql.core.data.dataItem.DataItem;
import com.songyang.yougnsql.core.transaction.TransactionManager;
import com.songyang.yougnsql.core.utils.ByteUtils;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

/**
 * @ClassName Node
 * @Description Node节点是b+树当中的节点抽象。Node结构如下：
 * [LeafFlag][KeyNumber][SiblingUid]
 * [Son0][Key0][Son1][Key1]...[SonN][KeyN]
 * 先将所有的数据添加到node节点，即b+树的所有叶子节点
 * 再根据叶子节点的数据对node进行分割为多个非叶子节点。
 * 其中非叶子节点的key就是key，son记录着叶子节点的uid（因为node节点也是文件存储，自然有uid）
 * 叶子节点的key就是key，son是数据的uid，根据兄弟节点的uid产生一条单向链表(最后一个叶子节点的key永远为整数的最大值，方便查找)，可以完成二分的查找。
 * key是被索引的列，如果是数字类型就是本身，如果是其他类型则会通过某些规则变成数字类型
 * @date 2022/12/22 13:39
 * @Author yanceysong
 * @Version 1.0
 */
public class Node {
    /**
     * 节点数据为空的引用
     */
    public static final int NODE_DATA_IS_NULL = -1;
    /**
     * son的长度
     */
    private static final int LEN_SON = 8;
    /**
     * len的长度
     */
    private static final int LEN_KEY = 8;
    /**
     * 兄弟节点的长度
     */
    private static final int LEN_SIBLING = 8;
    /**
     * 是否是叶子节点存储偏移量
     */
    private static final int IS_LEAF_OFFSET = 0;
    /**
     * keyNumber存储偏移量
     */
    private static final int KEY_NUMBER_OFFSET = IS_LEAF_OFFSET + 1;
    /**
     * 兄弟节点的存储偏移量
     */
    private static final int SIBLING_UID_OFFSET = KEY_NUMBER_OFFSET + 2;
    /**
     * 整个node节点的头部信息大小
     */
    private static final int NODE_HEADER_SIZE = SIBLING_UID_OFFSET + LEN_SIBLING;
    /**
     * 一个节点最多的key，son对数
     */
    private static final int BALANCE_NUMBER = 32;
    /**
     * 一个节点的大小
     */
    private static final int NODE_SIZE = NODE_HEADER_SIZE + (2 * 8) * (BALANCE_NUMBER * 2 + 2);
    /**
     * 叶子节点的标记位
     */
    private static final byte LEAF = 1;
    private BPlusTree bPlusTree;
    private DataItem dataItem;
    private byte[] raw;
    private long uid;

    public Node() {

    }

    /**
     * 设置数据为叶子节点
     *
     * @param raw    数据
     * @param isLeaf 是否是叶子节点
     */
    public static void setRawIsLeaf(byte[] raw, boolean isLeaf) {
        raw[IS_LEAF_OFFSET] = isLeaf ? LEAF : LEAF ^ 1;
    }

    /**
     * 检查数据是否是叶子节点
     *
     * @param raw 数据
     * @return 检查结果
     */
    public static boolean checkRawIfLeaf(byte[] raw) {
        return raw[IS_LEAF_OFFSET] == LEAF;
    }

    /**
     * 给数据设置KeyNumber
     *
     * @param raw       数据
     * @param keyNumber 要设置的keyNumber
     */
    private static void setKeyNumber(byte[] raw, int keyNumber) {
        System.arraycopy(ByteUtils.short2Byte((short) keyNumber), 0, raw, KEY_NUMBER_OFFSET, 2);
    }

    /**
     * 通过数据获取KeyNumber
     *
     * @param raw 数据
     * @return KeyNumber
     */
    private static int getKeyNumber(byte[] raw) {
        return ByteUtils.parseShort(Arrays.copyOfRange(raw, KEY_NUMBER_OFFSET, KEY_NUMBER_OFFSET + 2));
    }

    /**
     * 创建一个根节点的数据
     *
     * @param left  左子树的uid
     * @param right 右子树的uid
     * @param key   key
     * @return 根节点的数据
     */
    public static byte[] newRootRaw(long left, long right, long key) {
        byte[] raw = new byte[NODE_SIZE];
        setRawIsLeaf(raw, false);
        setKeyNumber(raw, 2);
        setSibling(raw, NODE_DATA_IS_NULL);
        setSon(raw, left, 0);
        setKey(raw, key, 0);
        setSon(raw, right, 1);
        setKey(raw, Long.MAX_VALUE, 1);
        return raw;
    }

    /**
     * 创建一个空的根节点数据
     *
     * @return 空的根节点数据
     */
    public static byte[] newNilRootRaw() {
        byte[] raw = new byte[NODE_SIZE];
        setRawIsLeaf(raw, true);
        setKeyNumber(raw, 0);
        setSibling(raw, NODE_DATA_IS_NULL);
        return raw;
    }

    /**
     * 这是兄弟节点的uid数据
     *
     * @param raw     数据
     * @param sibling 兄弟节点的uid
     */
    private static void setSibling(byte[] raw, long sibling) {
        System.arraycopy(ByteUtils.long2Byte(sibling), 0, raw, SIBLING_UID_OFFSET, LEN_SIBLING);
    }

    /**
     * 通过数据获取兄弟节点的uid
     *
     * @param raw 数据
     * @return 兄弟节点的uid
     */
    private static long getSibling(byte[] raw) {
        return ByteUtils.parseLong(Arrays.copyOfRange(raw, SIBLING_UID_OFFSET, SIBLING_UID_OFFSET + LEN_SIBLING));
    }

    /**
     * 给数据设置uid是他的第几个son
     *
     * @param raw    数据
     * @param uid    uid
     * @param number 第几个son
     */
    private static void setSon(byte[] raw, long uid, int number) {
        int offset = NODE_HEADER_SIZE + number * (LEN_KEY + LEN_SON);
        System.arraycopy(ByteUtils.long2Byte(uid), 0, raw, offset, LEN_SON);
    }

    /**
     * 根据数据获取它第几个son
     *
     * @param raw    数据
     * @param number 第几个son
     * @return son的uid
     */
    private static long getSon(byte[] raw, int number) {
        int offset = NODE_HEADER_SIZE + number * (LEN_KEY + LEN_SON);
        return ByteUtils.parseLong(Arrays.copyOfRange(raw, offset, offset + LEN_SON));
    }

    /**
     * 给数据设置uid是他的第几个sonKey
     *
     * @param raw    数据
     * @param key    uid
     * @param number 第几个son
     */
    private static void setKey(byte[] raw, long key, int number) {
        int offset = NODE_HEADER_SIZE + number * (LEN_KEY + LEN_SON) + LEN_SON;
        System.arraycopy(ByteUtils.long2Byte(key), 0, raw, offset, LEN_KEY);
    }

    /**
     * 根据数据获取它第几个sonKey
     *
     * @param raw    数据
     * @param number 第几个son
     * @return son的uid
     */
    private static long getKey(byte[] raw, int number) {
        int offset = NODE_HEADER_SIZE + number * (LEN_KEY + LEN_SON) + LEN_SON;
        return ByteUtils.parseLong(Arrays.copyOfRange(raw, offset, offset + LEN_KEY));
    }

    /**
     * 从一个节点的第几个子节点的数据复制到另一个节点
     *
     * @param from 源节点
     * @param to   移动到哪
     * @param kth  第几个子节点
     */
    private static void copyRawFromKth(byte[] from, byte[] to, int kth) {
        int offset = NODE_HEADER_SIZE + kth * (LEN_KEY + LEN_SON);
        //保留头部信息
        System.arraycopy(from, offset, to, NODE_HEADER_SIZE, from.length - offset);
    }

    /**
     * 把某一个位置开始的数据向后移动一个son和key的位置
     *
     * @param raw 数据
     * @param kth 位置
     */
    private static void shiftRawKth(byte[] raw, int kth) {
        //开始的偏移量
        int begin = NODE_HEADER_SIZE + (kth + 1) * (LEN_KEY + LEN_SON);
        //结束的偏移量位置
        int end = NODE_SIZE - 1;
        //还有空间
        if (end + 1 - begin >= 0)
            //全体向后移动
            System.arraycopy(raw, begin - (LEN_KEY + LEN_SON), raw, begin, end + 1 - begin);
    }

    /**
     * 通过node节点的uid加载一个node节点
     *
     * @param bTree b+树
     * @param uid   node节点的uid
     * @return node节点
     * @throws Exception 异常
     */
    static Node loadNode(BPlusTree bTree, long uid) throws Exception {
        DataItem dataItem = bTree.dataManager.read(uid);
        assert dataItem != null;
        Node node = new Node();
        node.bPlusTree = bTree;
        node.dataItem = dataItem;
        node.raw = dataItem.data();
        node.uid = uid;
        return node;
    }

    /**
     * 在非叶子节点当中查找key，返回下一层的结果
     *
     * @param key key
     * @return 查询的结果
     */
    public SearchNextRes searchNext(long key) {
        dataItem.rLock();
        try {
            //查到有几个key和son
            int keyNumber = getKeyNumber(raw);
            for (int i = 0; i < keyNumber; i++) {
                //非叶子节点的key记录
                long ik = getKey(raw, i);
                if (key < ik) {
                    //找到所在的范围了
                    return new SearchNextRes(getSon(raw, i), Node.NODE_DATA_IS_NULL);
                }
            }
            //如果没有找到，就去兄弟节点找（层次遍历）
            return new SearchNextRes(Node.NODE_DATA_IS_NULL, getSibling(raw));
        } finally {
            dataItem.rUnLock();
        }
    }

    /**
     * 在节点中范围查询
     *
     * @param leftKey  左子树的key
     * @param rightKey 右子树的key
     * @return 叶子节点的范围查找结果
     */
    public LeafSearchRangesRes leafSearchRange(long leftKey, long rightKey) {
        dataItem.rLock();
        ArrayList<Long> uids = new ArrayList<>();
        long sibling = NODE_DATA_IS_NULL;
        try {
            int keyNumbers = getKeyNumber(raw);
            for (int i = 0; i <= keyNumbers; i++) {
                long key = getKey(raw, i);
                //把这个范围内的uid全部保存下来
                if (key >= leftKey && key <= rightKey) {
                    uids.add(getSon(raw, i));
                } else if (key > rightKey) {
                    if (i == keyNumbers) {
                        //如果在范围外了，那么查看当前的次序是不是最后一个，如果是最后一个son和key，那么其兄弟节点可能还有
                        //记录下来，这个操作只需要执行一次
                        sibling = getSibling(raw);
                    }
                    break;
                }
            }
            return new LeafSearchRangesRes(uids, sibling);
        } finally {
            dataItem.rUnLock();
        }
    }

    /**
     * 向节点当中插入数据
     *
     * @param uid 资源的uid
     * @param key key
     * @return 插入结果
     */
    private boolean insert(long uid, long key) {
        //获取当前的keyNumbers
        int keyNumbers = getKeyNumber(raw);
        //遍历每一个key和son
        for (int i = 0; i <= keyNumbers; i++) {
            //获取到的key
            long getKey = getKey(raw, i);
            //如果获取的key大于当前的key，说明找到插入的位置了
            if (getKey >= key || getKey == 0) {
                //如果是叶子节点
                if (checkRawIfLeaf(raw)) {
                    //这个位置之后的数据移动一个节点
                    shiftRawKth(raw, i);
                    //留下来的位置给它
                    setKey(raw, key, i);
                    setSon(raw, uid, i);
                    setKeyNumber(raw, ++keyNumbers);
                } else {
                    //如果不是叶子节点
                    long k = getKey(raw, i);
                    //保留这个位置所指向的节点uid
                    setKey(raw, key, i);
                    //这个节点后面的数据向后移动
                    shiftRawKth(raw, ++i);
                    //把原来的数据添加进去
                    setKey(raw, k, ++i);
                    //因为这个key比当前的key小，当前的key替换了新的位置，将节点uid放在下一个区间
                    setSon(raw, uid, ++i);
                    setKeyNumber(raw, ++keyNumbers);
                }
                return true;
            }
            //如果查询完了所有的节点并且他的兄弟节点不为空，因为要插入的节点可能是下一个节点
            if (i == keyNumbers && getSibling(raw) != NODE_DATA_IS_NULL) return false;
        }
        return false;
    }

    /**
     * 分割节点
     *
     * @return 分割节点返回值
     */
    private SplitRes split() throws Exception {
        byte[] newRaw = new byte[NODE_SIZE];
        //都在一层，原来是叶子现在还是叶子节点
        setRawIsLeaf(newRaw, checkRawIfLeaf(raw));
        //设置keyNumber
        setKeyNumber(newRaw, BALANCE_NUMBER);
        //原来节点的兄弟节点给他
        setSibling(newRaw, getSibling(raw));
        //之后的数据全部赋值到新的数据域
        copyRawFromKth(raw, newRaw, BALANCE_NUMBER);
        //新数据插入到文件返回插入的uid
        long son = bPlusTree.dataManager.insert(TransactionManager.SUPER_XID, newRaw);
        setKeyNumber(raw, BALANCE_NUMBER);
        setSibling(raw, son);
        return new SplitRes(son, getKey(newRaw, 0));
    }

    /**
     * 插入，如果需要分割则会进行分割
     *
     * @param uid uid
     * @param key key
     * @return 插入节点返回值
     */
    public InsertAndSpiltRes insertAndSpilt(long uid, long key) {
        boolean success = false;
        dataItem.before();
        try {
            //把uid和key插入到当前的node节点
            success = insert(uid, key);
            //插入失败，应该把这个节点插入到当前层次的下一个节点
            if (!success) {
                return new InsertAndSpiltRes(getSibling(raw), NODE_DATA_IS_NULL, NODE_DATA_IS_NULL);
            }
            //插入成功判断是否需要分割
            if (needSplit()) {
                try {
                    //如果需要分割，那么分割返回第一个key,和son
                    SplitRes split = split();
                    return new InsertAndSpiltRes(NODE_DATA_IS_NULL, split.newSon, split.newKey);
                } catch (Exception e) {
                    ExceptionHandler.handler(e);
                }
            } else return new InsertAndSpiltRes(NODE_DATA_IS_NULL, NODE_DATA_IS_NULL, NODE_DATA_IS_NULL);
        } finally {
            if (success) {
                dataItem.after(TransactionManager.SUPER_XID);
            } else dataItem.unBefore();
        }
        return null;
    }

    /**
     * 是否需要进行分割
     *
     * @return 结果
     */
    private boolean needSplit() {
        return BALANCE_NUMBER * 2 == getKeyNumber(raw);
    }

    /**
     * node节点释放
     */
    public void release() {
        dataItem.release();
    }

    /**
     * 判断当前节点的数据是不是叶子节点
     *
     * @return 是否是叶子节点
     */
    public boolean isLeaf() {
        dataItem.rLock();
        try {
            return checkRawIfLeaf(raw);
        } finally {
            dataItem.rUnLock();
        }
    }

    @Override
    public String toString() {
        StringBuilder sb = new StringBuilder();
        sb.append("Is leaf: ").append(checkRawIfLeaf(raw)).append("\n");
        int KeyNumber = getKeyNumber(raw);
        sb.append("KeyNumber: ").append(KeyNumber).append("\n");
        sb.append("sibling: ").append(getSibling(raw)).append("\n");
        for (int i = 0; i < KeyNumber; i++) {
            sb.append("son: ")
                    .append(getSon(raw, i))
                    .append(", key: ")
                    .append(getKey(raw, i))
                    .append("\n");
        }
        return sb.toString();
    }

    protected static class SearchNextRes {
        public long uid;
        public long siblingUid;

        public SearchNextRes(long uid, long siblingUid) {
            this.uid = uid;
            this.siblingUid = siblingUid;
        }
    }

    protected static class LeafSearchRangesRes {
        public List<Long> uids;
        public long siblingUid;

        public LeafSearchRangesRes(List<Long> uids, long siblingUid) {
            this.uids = uids;
            this.siblingUid = siblingUid;
        }
    }

    protected static class InsertAndSpiltRes {
        public long siblingUid;
        public long newSon;
        public long newKey;

        public InsertAndSpiltRes(long siblingUid, long newSon, long newKey) {
            this.siblingUid = siblingUid;
            this.newSon = newSon;
            this.newKey = newKey;
        }
    }

    protected static class SplitRes {
        public long newSon;
        public long newKey;

        public SplitRes(long newSon, long newKey) {
            this.newSon = newSon;
            this.newKey = newKey;
        }
    }
}
